package com.example.toy.core.reader;

import com.example.toy.core.dataset.BigCollection;
import com.example.toy.core.engine.EngineContext;
import com.example.toy.core.engine.GlobalEngineContext;
import com.example.toy.core.filter.Filters;
import com.example.toy.core.filter.ToyFilter;
import com.example.toy.core.record.ToyRecord;
import com.example.toy.core.record.key.ToyRecordKey;
import com.example.toy.core.table.ToyTableMeta;
import com.example.toy.core.utils.FilterHelper;
import com.example.toy.core.utils.RecordKeyHelper;
import lombok.Data;
import org.apache.hadoop.fs.Path;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;

@Data
public class DefaultToyTableReader implements ToyTableReader {
  private ToyTableMeta tableMeta;

  public DefaultToyTableReader(ToyTableMeta tableMeta) {
    this.tableMeta = tableMeta;
  }

  @Override
  public Optional<ToyRecord> readKey(ToyRecordKey key) {
    BigCollection<ToyRecord> collection = readKeys(Arrays.asList(key));
    return collection.collect().stream().findAny();
  }

  @Override
  public BigCollection<ToyRecord> readKeys(List<ToyRecordKey> keys) {
    Optional<ToyFilter> toyFilter =
        keys.stream().map(RecordKeyHelper::toToyFilter).reduce(ToyFilter::or);
    return readFull(toyFilter.orElse(Filters.ALWAYS_TRUE));
  }

  @Override
  public BigCollection<ToyRecord> readFull(ToyFilter filter) {
    Map<String, Path> partitionPathMap = tableMeta.getPartitionPathMap();

    ToyFilter partitionFilter =
        FilterHelper.findPartitionFilter(tableMeta.getPartitionSchema(), filter);

    List<String> partitions =
        partitionPathMap.keySet().stream()
            //            .filter(partitionFilter::accept)
            .collect(Collectors.toList());

    List<BigCollection<ToyRecord>> collect =
        partitions.stream()
            .parallel()
            .map(p -> readPartition(p, filter))
            .collect(Collectors.toList());

    EngineContext engineContext = GlobalEngineContext.get();

    BigCollection<ToyRecord> collection = collect.stream()
        .reduce(BigCollection::union)
        .orElse(engineContext.empty());
    return collection;
  }

  /**
   * 读取某个分区下的表数据文件
   *
   * @param partition
   * @param filter
   * @return
   */
  @Override
  public BigCollection<ToyRecord> readPartition(String partition, ToyFilter filter) {
    ToyPartitionReader partitionReader = new DefaultToyPartitionReader(tableMeta, partition);
    return partitionReader.read(filter);
  }
}
